package org.apache.rocketmq.common.thread;

import com.google.common.collect.Lists;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.utils.ThreadUtils;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.*;

/**
 * @方法描述：线程池监控组件，在proxy模块中，在源码中，只要是RemotingProtocolServer对象中用到的线程池，基本上都是由这个组件创建的
 */
public class ThreadPoolMonitor {
    //记录线程栈的日志记录器
    private static Logger jstackLogger = LoggerFactory.getLogger(ThreadPoolMonitor.class);
    //记录每一个执行器水位信息的日志记录器，所谓水位信息其实就是执行器执行任务的慢速时间
    private static Logger waterMarkLogger = LoggerFactory.getLogger(ThreadPoolMonitor.class);
    //记录每一个执行器水位信息的日志记录器，所谓水位信息其实就是执行器执行任务的慢速时间
    private static final List<ThreadPoolWrapper> MONITOR_EXECUTOR = new CopyOnWriteArrayList<>();
    //定时任务线程池
    private static final ScheduledExecutorService MONITOR_SCHEDULED =
            ThreadUtils.newSingleThreadScheduledExecutor(new ThreadFactoryBuilder()
                    .setNameFormat("ThreadPoolMonitor-%d").build());
    //日志记录线程池状态信息的时间间隔
    private static volatile long threadPoolStatusPeriodTime = TimeUnit.SECONDS.toMillis(3);
    //否开启记录线程栈信息的标志
    private static volatile boolean enablePrintJstack = true;
    //记录线程栈信息的时间间隔
    private static volatile long jstackPeriodTime = 60000;
    //最新一次记录线程栈信息的时间
    private static volatile long jstackTime = System.currentTimeMillis();

    public static void config(Logger jstackLoggerConfig, Logger waterMarkLoggerConfig,
                              boolean enablePrintJstack, long jstackPeriodTimeConfig, long threadPoolStatusPeriodTimeConfig) {
        jstackLogger = jstackLoggerConfig;
        waterMarkLogger = waterMarkLoggerConfig;
        threadPoolStatusPeriodTime = threadPoolStatusPeriodTimeConfig;
        ThreadPoolMonitor.enablePrintJstack = enablePrintJstack;
        jstackPeriodTime = jstackPeriodTimeConfig;
    }

    //创建线程池的方法
    public static ThreadPoolExecutor createAndMonitor(int corePoolSize,
                                                      int maximumPoolSize,
                                                      long keepAliveTime,
                                                      TimeUnit unit,
                                                      String name,
                                                      int queueCapacity) {
        return createAndMonitor(corePoolSize, maximumPoolSize,
                keepAliveTime, unit, name, queueCapacity, Collections.emptyList());
    }

    public static ThreadPoolExecutor createAndMonitor(int corePoolSize,
                                                      int maximumPoolSize,
                                                      long keepAliveTime,
                                                      TimeUnit unit,
                                                      String name,
                                                      int queueCapacity,
                                                      ThreadPoolStatusMonitor... threadPoolStatusMonitors) {
        return createAndMonitor(corePoolSize, maximumPoolSize, keepAliveTime, unit, name, queueCapacity,
                Lists.newArrayList(threadPoolStatusMonitors));
    }

    private static ThreadPoolExecutor createAndMonitor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                                                       String name, int queueCapacity, List<ThreadPoolStatusMonitor> threadPoolStatusMonitors) {

        ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) ThreadUtils.newThreadPoolExecutor(
                corePoolSize,
                maximumPoolSize,
                keepAliveTime,
                unit,
                new LinkedBlockingQueue<>(queueCapacity),
                new ThreadFactoryBuilder().setNameFormat(name + "-%d").build(),
                new ThreadPoolExecutor.DiscardOldestPolicy());

        //这里创建了一个线程池状态监听器的集合
        List<ThreadPoolStatusMonitor> printers = Lists.newArrayList(new ThreadPoolQueueSizeMonitor(queueCapacity));
        //把外层方法传递进来的监听器集合添加到集合中
        printers.addAll(threadPoolStatusMonitors);
        //在这里对线程池做了一层包装，创建了一个ThreadPoolWrapper对象，然后把真正的线程池和线程池监听器都封装到这个ThreadPoolWrapper对象中了
        MONITOR_EXECUTOR.add(ThreadPoolWrapper.builder()
                .name(name)
                .threadPoolExecutor(threadPoolExecutor)
                .statusPrinters(printers)
                .build());


        return threadPoolExecutor;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/31
     * @方法描述：记录线程池状态信息的方法
     */
    public static void logThreadPoolStatus() {
        //遍历所有的线程池包装器
        for (ThreadPoolWrapper threadPoolWrapper : MONITOR_EXECUTOR) {
            //从当前正在遍历的线程池包装器对象中取出该线程池的监听器集合
            List<ThreadPoolStatusMonitor> monitors = threadPoolWrapper.getStatusPrinters();
            //遍历该线程池的监听器集合
            for (ThreadPoolStatusMonitor monitor : monitors) {
                //执行每一个监听器的value方法，获取到线程池的相关信息
                //在第三版本代码中，这里得到的就是topicRouteExecutor执行器处理根据主题获取路由信息的任务的慢速时间
                //topicRouteExecutor执行器在RemotingProtocolServer对象中被创建
                double value = monitor.value(threadPoolWrapper.getThreadPoolExecutor());
                //日志记录器记录线程池状态信息，这些信息包括执行器的名称，监听器的描述信息，以及监听器的value值
                waterMarkLogger.info("\t{}\t{}\t{}", threadPoolWrapper.getName(), monitor.describe(), value);
                //判断是否启用了记录线程栈信息的功能
                if (enablePrintJstack) {
                    //如果启动了那就从时间上判断一下是否可以记录线程栈信息，monitor.needPrintJstack()这个方法是用来判断线程池队列第一个任务是否超时了
                    //也就是是等待被执行的时间是否达到了慢速时间的阈值，超过了则意味着需要记录线程栈信息
                    if (monitor.needPrintJstack(threadPoolWrapper.getThreadPoolExecutor(), value) &&
                            System.currentTimeMillis() - jstackTime > jstackPeriodTime) {
                        //如果可以就更新最新一次记录线程栈信息的时间，并记录线程栈信息
                        jstackTime = System.currentTimeMillis();
                        jstackLogger.warn("jstack start\n{}", UtilAll.jstack());
                    }
                }
            }
        }
    }


    public static void init() {
        MONITOR_SCHEDULED.scheduleAtFixedRate(
                ThreadPoolMonitor::logThreadPoolStatus, threadPoolStatusPeriodTime, threadPoolStatusPeriodTime, TimeUnit.MILLISECONDS);
    }


    public static void shutdown() {
        MONITOR_SCHEDULED.shutdown();
    }
}
